/**    
 * @Title: MemoryChannel.java  
 * @Package cn.com.ytst.data.integration.core.channel  
 * @Description: TODO(用一句话描述该文件做什么)  
 * @author yanhewei@vv.cc.com    
 * @date 2020年7月5日 下午6:41:26  
 * @version V1.0    
*/
package cn.com.ytst.data.integration.core.channel;

import java.util.Collection;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**  
 * @ClassName: MemoryChannel  
 * @Description: TODO(这里用一句话描述这个类的作用)  
 * @author yanhewei@vv.cc.com  
 * @date 2020年7月5日 下午6:41:26  
 *    
*/
public class MemoryChannel implements PipleChannel{

	private ArrayBlockingQueue<String> queue = null;

    private AtomicLong memoryBytes = null;

    private ReentrantLock lock = new ReentrantLock();
    private Condition empty ;//写锁

    private Condition notEmpty ;//读锁

    private int queuesize = 10000;

    private int initcount = 0;

    public  MemoryChannel() {

        queue = new ArrayBlockingQueue<>(queuesize);
        memoryBytes = new AtomicLong(initcount);
        empty = lock.newCondition();//写锁
        notEmpty = lock.newCondition();//读锁
    }

    //往里面放入数据
    @Override
    public void doPush(String record) {
    	
        try {
			queue.put(record);
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}
    }

    //批量插入数据
    @Override
    public void doPushAll(Collection<String> records) {

    }




    //拉取数据
    @Override
    public String doPull() {
    	try {
			
			String r = this.queue.take();
			return r;
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
			throw new IllegalStateException(e);
		}
    }

    @Override
    public boolean isEmpty() {
        return this.queue.isEmpty();
    }

    @Override
    public int size() {
        return this.queue.size();
    }

    @Override
    public void readflush() {
        notEmpty.signal();
    }


}
